package io.gitee.hefren.hhmq.store;

import com.google.common.base.MoreObjects;
import com.google.common.base.Strings;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import io.gitee.hefren.hhmq.model.HhMessage;
import io.gitee.hefren.hhmq.server.HhMQ;

import java.io.File;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.Objects;

/**
 * 内存文件映射，用于替换 内存队列
 * @Date 2024/7/31
 * @Author lifei
 */
public class HhStore {

    private final String topic;
    public final static int LEN = 1024 * 1024;
    private static final int MESSAGE_BYTES_LEN_MAX_NUM = 10;
    private static final String NUM_REGEX = "^[0-9]+$";
    private static final char PAD_CHAR = '0';

    private MappedByteBuffer mappedByteBuffer;
    private static final Gson gson = new Gson();
    public HhStore(String topic) {
        this.topic = topic;
    }



    // 初始化， 获取MappedByteBuffer
    public void init() {
        File file = createTopicFile();
        Path path = Paths.get(file.getAbsolutePath());
        try (FileChannel fileChannel = (FileChannel) Files.newByteChannel(path, StandardOpenOption.READ, StandardOpenOption.WRITE)) {
            this.mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, LEN);
            // 加载topic的数据文件，把写偏移量移动到正确的位置
            int offset = loadTopicDataFile();
            this.mappedByteBuffer.position(offset);
        }catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    // 加载topic的数据文件，获取到合适的写偏移量位置
    private int loadTopicDataFile() {
        int offset = 0;
        ByteBuffer byteBuffer = mappedByteBuffer.asReadOnlyBuffer();
        while (true) {
            byteBuffer.position(offset);
            // 获取消息头信息
            byte[] bytes = new byte[MESSAGE_BYTES_LEN_MAX_NUM];
            byteBuffer.get(bytes);
            String message = new String(bytes, StandardCharsets.UTF_8);
            if (!message.matches(NUM_REGEX)) {
                break;
            }
            // 获取消息体的长度
            int messageBodyLen = Integer.parseInt(message);
            if (messageBodyLen<=0) {
                break;
            }
            int messageLen = MESSAGE_BYTES_LEN_MAX_NUM + messageBodyLen;
            HhIndexer.addEntry(topic, offset, messageLen);
            offset += messageLen;
        }
        return offset;
    }

    // 写消息
    public void write(HhMessage<?> hhMessage) {
        if (Objects.isNull(hhMessage)) {
            return;
        }
        String message = gson.toJson(hhMessage);
        // 对message进行编码： 添加消息头
        message = encodeMessage(message);
        byte[] bytes = message.getBytes(StandardCharsets.UTF_8);
        // 获取写消息的偏移偏移量
        int offset = mappedByteBuffer.position();
        // 获取消息的字节长度
        int len = bytes.length;
        // 保存offset 和 len的关系
        HhIndexer.addEntry(topic, offset, len);
        mappedByteBuffer.put(bytes);
    }

    // 为消息添加消息头，消息头是消息字节的长度
    // 假设一条消息最大长度是1G： 1 * 1024 * 1024 * 1024 =》 byte
    // 字节最大长度为： 10_0000_0000
    // 把message转化成byte[] , 这个byte[] 的长度可以用10位数表示
    private String encodeMessage(String message) {
        byte[] bytes = message.getBytes(StandardCharsets.UTF_8);
        int len = bytes.length;
        // 用10位表示长度
        String messageHeader = Strings.padStart(String.valueOf(len), MESSAGE_BYTES_LEN_MAX_NUM, PAD_CHAR);
        return messageHeader + message;
    }

    // 对消息解码： 去除消息头
    private String decodeMessage(String message) {
        return message.substring(MESSAGE_BYTES_LEN_MAX_NUM);
    }

    // 读消息
    public HhMessage<?> read(int offset) {
        if (offset<0 || offset>=LEN) {
            return null;
        }
        // 获取Entry
        HhIndexer.Entry entry = HhIndexer.getEntry(topic, offset);
        if (Objects.isNull(entry)) {
            return null;
        }
        ByteBuffer byteBuffer = mappedByteBuffer.asReadOnlyBuffer();
        // 定位到要开始读取的位置
        byteBuffer.position(entry.getOffset());
        // 读取消息的长度
        byte[] bytes = new byte[entry.getLen()];
        byteBuffer.get(bytes);
        // 转换成字符串
        String message = new String(bytes, StandardCharsets.UTF_8);
        // 对消息解码：去除消息头
        message = decodeMessage(message);
        TypeToken<HhMessage<?>> typeToken = new TypeToken<>(){};
        return gson.fromJson(message, typeToken.getType());

    }



    private File createTopicFile() {
        try {
            String fileName = Strings.lenientFormat("%s.bat", topic);
            File file = new File(fileName);
            if (!file.exists()) {
                boolean newFile = file.createNewFile();
                System.out.println(Strings.lenientFormat("===> %s, create file: %s", newFile, fileName));
            }
            return file;
        }catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取消息的偏移量
     * @return
     */
    public int getMessageOffset() {
        return mappedByteBuffer.position();
    }

    public String getTopic() {
        return topic;
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(HhStore.class)
                .add("topic", topic)
                .toString();
    }


}
